package spring.cloud.simple.config;

import java.util.List;
import java.util.ArrayList;
import java.util.Map;
import java.util.HashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;
import javax.annotation.PostConstruct;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Configuration;

import org.I0Itec.zkclient.IZkChildListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.transport.TSocket;
import org.apache.thrift.transport.TTransportException;

import spring.cloud.simple.interfaces.UserService;

@Configuration
public class ZooKeeperConfig {
    @Value("${service.name}")
    private String serviceName;

    @Value("${zookeeper.server.list}")
    private String zookeeperList;

    private ExecutorService executor = Executors.newSingleThreadExecutor();

    public static Map<String, UserService.Client> serviceMap = new HashMap<String, UserService.Client>();

    @PostConstruct
    public void init() {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                startZookeeper();
                try {
                    Thread.sleep(1000 * 60 * 60 * 24 * 360 * 10);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        });
    }

    private void startZookeeper() {
        List<String> children = new ArrayList<String>();
        String servicePath = "/" + serviceName;
        ZkClient zkClient  = new ZkClient(zookeeperList);
        boolean serviceExists = zkClient.exists(servicePath);
        if (serviceExists) {
            children = zkClient.getChildren(servicePath);
        } else {
            throw new RuntimeException("service does not exist!");
        }

        for (String instanceName : children) {
            if (!serviceMap.containsKey(instanceName)) {
                serviceMap.put(instanceName, createUserService(instanceName));
            }
        }

        zkClient.subscribeChildChanges(servicePath, new IZkChildListener() {
            @Override
            public void handleChildChange(String parentPath, List<String> children) throws Exception {
                for (String instanceName : children) {
                    if (!serviceMap.containsKey(instanceName)) {
                        serviceMap.put(instanceName, createUserService(instanceName));
                    }
                }
                for (Map.Entry<String, UserService.Client> entry : serviceMap.entrySet()) {
                    if (!children.contains(entry.getKey())) {
                        UserService.Client client = serviceMap.get(entry.getKey());
                        try {
                            client.getInputProtocol().getTransport().close();
                            client.getOutputProtocol().getTransport().close();
                        } catch (Exception e) {
                            e.printStackTrace();
                        }
                        serviceMap.remove(entry.getKey());
                    }
                }
                System.out.println(parentPath + " event activate!");
            }
        });
    }

    private UserService.Client createUserService(String serviceInstanceName) {
        String ip = serviceInstanceName.split("-")[1];
        TSocket tSocket = new TSocket(ip, 7911);
        try {
            tSocket.open();
        } catch (TTransportException e) {
            e.printStackTrace();
        }
        return new UserService.Client(new TBinaryProtocol(tSocket));
    }
}
